package org.jeecg.modules.cascv.kafka;

import cn.hutool.core.util.RandomUtil;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.jeecg.modules.cascv.kafka.listener.KafkaListeners;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

/**
 * @author cwenao
 * @version $Id KafkaConsumerConfig.java, v 0.1 2017-01-21 21:11 cwenao Exp $$
 */
@Configuration
@EnableKafka
@ConditionalOnProperty(
        value = {"bootcwenao.kafka.enable"},
        matchIfMissing = true
)
public class KafkaConsumerConfig {

    @Value("${bootcwenao.kafka.binder.brokers}")
    private String brokers;

    @Value("${bootcwenao.kafka.group}")
    private String group;

    @Value("${bootcwenao.kafka.consumer.auto-commit-interval}")
    private String autoCommitInterval;

    @Value("${bootcwenao.kafka.consumer.session-timeout}")
    private String sessionTimeout;

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(10);
        factory.getContainerProperties().setPollTimeout(4000);

        return factory;
    }

    @Bean
    public KafkaListeners kafkaListeners() {
        return new KafkaListeners();
    }

    public ConsumerFactory<String, String> consumerFactory() {

        Map<String, Object> properties = new HashMap<>();
        //消费者客户端的id
        properties.put(ConsumerConfig.CLIENT_ID_CONFIG, "CLIENT_C" + RandomUtil.randomInt(100));
        //连接集群broker地址
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
        //是否开启自动提交消费位移的功能，默认true
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        //自动提交消费位移的时间间隔，默认5000ms
        properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
        //检测消费者是否失效的超时时间，默认10000ms
        properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
        //key反序列化方式
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //value反序列化方式
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        //消费者所属消费组的唯一标识
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, group + RandomUtil.randomString(3));
        //如果分区没有初始偏移量，或者当前偏移量服务器上不存在时，将使用的偏移量设置，earliest从头开始消费，latest从最近的开始消费，none抛出异常
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
//        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "com.cascv.device.kafka.DecodeingKafka");

        return new DefaultKafkaConsumerFactory<>(properties);
    }
}